package spring.cloud.tasks.executor.executor;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.executor.task.TaskExecutor;
import spring.cloud.tasks.executor.task.statistics.ProcessCountStatistics;
import spring.cloud.tasks.executor.utils.log.LogEvents;
import spring.cloud.tasks.executor.utils.log.LogUtils;

import java.util.Iterator;
import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 执行统计信息清零的动作
 */
@Slf4j
@Setter
@Getter
public class ExecutorProcessCountResetTask extends TimerTask {
    private String executorId;

    public ExecutorProcessCountResetTask(String executorId) {
        this.executorId = executorId;
    }

    @Override
    public void run() {
        try {
            Map<String, ConcurrentHashMap<String, TaskExecutor>> executorIdToTaskIdToTaskExecutorMapMap = ExecutorIdToTaskIdToTaskExecutorMapMap.getExecutorIdToTaskIdToTaskExecutorMapMap();
            if (executorIdToTaskIdToTaskExecutorMapMap == null) {
                return;
            }

            // 只清零本executor的统计信息数据
            if (executorIdToTaskIdToTaskExecutorMapMap.containsKey(executorId)) {
                ConcurrentHashMap<String, TaskExecutor> taskIdToTaskExecutorMap = executorIdToTaskIdToTaskExecutorMapMap.get(executorId);
                if (taskIdToTaskExecutorMap == null) {
                    return;
                }
                Iterator<Map.Entry<String, TaskExecutor>> iterator = taskIdToTaskExecutorMap.entrySet().iterator();
                while (iterator.hasNext()) {
                    Map.Entry<String, TaskExecutor> next = iterator.next();
                    String taskId = next.getKey();
                    TaskExecutor taskExecutor = next.getValue();
                    // 清零内存统计值
                    ProcessCountStatistics.resetSuccessFailureCount(taskId, executorId);
                    // 清零zk统计值
                    taskExecutor.getTaskExecutorInfoService().persistProcessFailureCount(0);
                    taskExecutor.getTaskExecutorInfoService().persistProcessSuccessCount(0);
                    LogUtils.info(log, taskId, "{} reset the job {}'s statistics data", executorId, taskId);
                }
            }
        } catch (Throwable t) {
            LogUtils.error(log, LogEvents.ExecutorEvent.COMMON, "process count reset error", t);
        }
    }
}
